/*
 * 文 件 名:  JavaSever.java
 * 版    权:  Archermind.NANJING 2014-9-3,  All rights reserved
 * 描    述:  <描述>
 * 修 改 人:  Administrator
 * 修改时间:  2014-9-3   上午11:20:27
 * 跟踪单号:  <跟踪单号>
 * 修改单号:  <修改单号>
 * 修改内容:  <修改内容>
 */
package com.demo.netty.mqtt;

import com.demo.netty.util.Config;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

import java.sql.Timestamp;

/**
 * MQTT JAVA 接收和publish 端
 * 
 * @author  lpf
 * @version  [版本号, 2014-9-3]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class MqttJavaSever implements MqttCallback, Runnable
{
    private static final Logger logger = Logger.getLogger(MqttJavaSever.class);
    
	public static final String MSGEND = "]";
    
	public static final String MSGHEAD = "[";
	
	public static final String RESULTEND = "}]";
    
	public static final String RESULTHEAD = "[{";
    
	public static final String CHARSET = "UTF-8";
	
    // 订阅号 关键字
    public static String topic = Config.getProperty("mqtt_clientId");
    
    private String tmpDir = System.getProperty("java.io.tmpdir");
    
    // 接收broker 的缓存数据 false  true 清空缓存数据
    private boolean clean = false;
    
    private static String userName;
    
    private static String passWord;
    
    private static String brokerUrl;
    
    private static String clientId;
    
    private Object waiter = new Object();
    
    private MqttAsyncClient client = null;
    
    private MqttConnectOptions conOpt = null;
    
    private boolean state = false;
    
    private String[] topics;
    
    private int[] qoss;
    
    //private ExecutorService executorService = Executors.newFixedThreadPool(20);
    
    public MqttJavaSever()
    {
        topic = Config.getProperty("mqtt_topic");
        clean = Boolean.parseBoolean(Config.getProperty("mqtt_clean"));
        userName = Config.getProperty("mqtt_userName");
        passWord = Config.getProperty("mqtt_passWord");
        brokerUrl = Config.getProperty("mqtt_brokerUrl");
        clientId = Config.getProperty("mqtt_clientId");
        //qos = Integer.valueOf(ConfUtil.getInitProperty("qos"));
    }
    
    public void initConn()
    {
        
        // Construct the MqttClient instance
        try
        {
            MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(
                    tmpDir);
            conOpt = new MqttConnectOptions();
            // 心跳时间 30s
            conOpt.setKeepAliveInterval(60);
            conOpt.setCleanSession(clean);
            if (passWord != null)
            {
                conOpt.setPassword(this.passWord.toCharArray());
            }
            if (userName != null)
            {
                conOpt.setUserName(this.userName);
            }
            client = new MqttAsyncClient(this.brokerUrl, clientId, dataStore);
            // Set this wrapper as the callback handler
            client.setCallback(this);
            MqttConnector con = new MqttConnector();
            con.doConnect();
            
            this.topics = new String[] { topic };
            this.qoss = new int[] { 2 };
            
            while (true)
            {
                if (state)
                {
                    Subscriber sub1 = new Subscriber();
                    sub1.doSubscribe(topics, qoss);
                    break;
                }
                
                waitForStateChange(200);
            }
        }
        catch (MqttException e)
        {
            // TODO Auto-generated catch block
        	logger.error(e.getMessage());
        }
        
        //Publisher pub = new Publisher();
        // pub.doPublish(topic, qos, msg.getBytes());
    }
    
    /**
     * Wait for a maximum amount of time for a state change event to occur
     * @param maxTTW  maximum time to wait in milliseconds
     * @throws MqttException
     */
    private void waitForStateChange(int maxTTW) throws MqttException
    {
        synchronized (waiter)
        {
            
            try
            {
                waiter.wait(maxTTW);
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
            
        }
    }
    
    /**
     * Connect in a non-blocking way and then sit back and wait to be
     * notified that the action has completed.
     */
    public class MqttConnector
    {
        
        public MqttConnector()
        {
        }
        
        public void doConnect()
        {
            // Connect to the server
            // Get a token and setup an asynchronous listener on the token which
            // will be notified once the connect completes
            //log("Connecting to " + brokerUrl + " with client ID "
            //        + client.getClientId());
            
            IMqttActionListener conListener = new IMqttActionListener()
            {
                public void onSuccess(IMqttToken asyncActionToken)
                {
                    //log("Connected");
                    logger.info("conntion ok.");
                    //state = CONNECTED;
                    state = true;
                    carryOn();
                }
                
                public void onFailure(IMqttToken asyncActionToken,
                        Throwable exception)
                {
                    //ex = exception;
                    logger.error("MqttJavaSever MqttConnector conn error.");
                    logger.error(exception.getMessage());
                    state = false;
                    //log("connect failed" + exception);
                    //carryOn();
                }
                
                public void carryOn()
                {
                    synchronized (waiter)
                    {
                        //donext = true;
                        waiter.notifyAll();
                    }
                }
            };
            
            try
            {
                // Connect using a non-blocking connect
                client.connect(conOpt, "Connect sample context", conListener);
            }
            catch (MqttException e)
            {
                // If though it is a non-blocking connect an exception can be
                // thrown if validation of parms fails or other checks such
                // as already connected fail.
                //state = ERROR;
                //donext = true;
                // ex = e;
                logger.error(e.getMessage());
            }
        }
    }
    
    /**
     * Subscribe in a non-blocking way and then sit back and wait to be
     * notified that the action has completed.
     */
    public class Subscriber
    {
        public void doSubscribe(String[] strs, int... qos)
        {
            
            // Make a subscription
            // Get a token and setup an asynchronous listener on the token which
            // will be notified once the subscription is in place.
            //log("Subscribing to topic \"" + topicName + "\" qos " + qos);
            
            IMqttActionListener subListener = new IMqttActionListener()
            {
                public void onSuccess(IMqttToken asyncActionToken)
                {
                    //log("Subscribe Completed");
                    //state = SUBSCRIBED;
                    logger.info("onSuccess");
                    carryOn();
                }
                
                public void onFailure(IMqttToken asyncActionToken,
                        Throwable exception)
                {
                    //ex = exception;
                    //state = ERROR;
                    //log("Subscribe failed" + exception);
                    logger.info("onFailure");
                    carryOn();
                }
                
                public void carryOn()
                {
                    synchronized (waiter)
                    {
                        waiter.notifyAll();
                    }
                }
            };
            
            try
            {
                //                client.subscribe(topicName,
                //                        qos,
                //                        "Subscribe sample context",
                //                        subListener);
                client.subscribe(strs, qos, "subscribe params", subListener);
            }
            catch (MqttException e)
            {
                //state = ERROR;
                //donext = true;
                //ex = e;
            	logger.error(e.getMessage());
            }
        }
    }
    
    // 断网重连
    @Override
    public void connectionLost(Throwable arg0)
    {
        try
        {
            state = false;
            while (true)
            {
                //System.out.println("reconn start...");
                MqttConnector con = new MqttConnector();
                con.doConnect();
                
                if (state)
                {
                    //System.out.println(state + "   start subscriber.");
                    Subscriber sub = new Subscriber();
                    sub.doSubscribe(topics, qoss);
                    break;
                }
                
                waitForStateChange(5000);
            }
        }
        catch (Exception e)
        {
        	logger.error(e.getMessage());
            // TODO: handle exception
        }
    }
    
    @Override
    public void deliveryComplete(IMqttDeliveryToken arg0)
    {
        // TODO Auto-generated method stub
        
    }
    
    @Override
    public void messageArrived(String arg0, MqttMessage arg1) throws Exception
    {
        String time = new Timestamp(System.currentTimeMillis()).toString();
        String msg = new String(arg1.getPayload());
        try
        {
            if (null == msg || "".equals(msg.trim()))
            {
                return;
            }
            
        }
        catch (Exception e)
        {
            logger.error("error", e);
        }
        
    }
    
    @Override
    public void run()
    {
        try
        {
            logger.info("init mqtt subscript start...");
            
            initConn();
            
        }
        catch (Exception e)
        {
        	logger.error(e.getMessage());
            // TODO: handle exception
        }
        
        //System.exit(0);
    }
}
